package cn.doitedu.cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/8/31
 * @Desc: 学大数据，上多易教育
 *   每 5分钟，每类 页面中访问人数最多的前10个页面及其访问人数
 *
 *
 **/
public class Job3_TopUvPage_Per5Min {

    public static void main(String[] args) {


        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中的dwd层的行为日志明细
        tenv.executeSql(
                "  CREATE TABLE dwd_events_kafka(                   "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     user_id           BIGINT,                     "
                        +"      page_url          STRING,                     "
                        +"      page_type         STRING,                     "
                        + "     rt AS to_timestamp_ltz(event_time,3),         " // 表达式字段
                        + "     WATERMARK FOR  rt AS  rt -  INTERVAL '0' SECOND  "  // 声明watermark，目标字段就具备了时间属性，不再是一个普通字段
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',         "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 建表，映射mysql中的结果表
        tenv.executeSql(
                " create table dash_topn_page_per5min_mysql(     " +
                        "    window_start timestamp(3),            " +
                        "    window_end timestamp(3),              " +
                        "    page_type STRING,                        " +
                        "    page_url STRING,                   " +
                        "    uv_cnt bigint,                        " +
                        "    rn bigint,                        " +
                        "    PRIMARY KEY (window_start,window_end) NOT ENFORCED     " +
                        " ) with (                                      " +
                        "  'connector' = 'jdbc',                            " +
                        "  'url' = 'jdbc:mysql://doitedu:3306/doit40',  " +
                        "  'table-name' = 'dash_topn_page_per5min',             " +
                        "  'username' = 'root',                             " +
                        "  'password' = 'root'                              " +
                        " )                                                 "
        );


        tenv.executeSql(
                "insert into  dash_topn_page_per5min_mysql  " +
                        "WITH tmp AS (                                                           "+
                        "     SELECT                                                              "+
                        "        *                                                                "+
                        "     FROM  dwd_events_kafka                                              "+
                        "     WHERE event_id = 'page_load'                                        "+
                        " )                                                                       "+
                        "                                                                         "+
                        " SELECT                                                                  "+
                        "     window_start,                                                       "+
                        "     window_end,                                                         "+
                        "     page_type,                                                          "+
                        "     page_url,                                                           "+
                        "     uv,                                                                 "+
                        " 	rn                                                                    "+
                        " FROM (                                                                  "+
                        "     SELECT                                                              "+
                        "             window_start,                                               "+
                        "         	window_end,                                                   "+
                        "         	page_type,                                                    "+
                        "         	page_url,                                                     "+
                        "     		uv,                                                           "+
                        "     		row_number() over(                                            "+
                        " 			      PARTITION BY window_start,window_end,page_type          "+
                        " 				  ORDER BY uv DESC) as rn                                 "+
                        "     FROM (                                                              "+
                        "                                                                         "+
                        "         SELECT                                                          "+
                        "             window_start,                                               "+
                        "         	window_end,                                                   "+
                        "         	page_type,                                                    "+
                        "         	page_url,                                                     "+
                        "         	count(distinct user_id) as uv                                 "+
                        "         FROM TABLE(                                                     "+
                        "             TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '5' MINUTE)        "+
                        "         )                                                               "+
                        "         GROUP BY                                                        "+
                        "             window_start,                                               "+
                        "         	window_end,                                                   "+
                        "         	page_type,                                                    "+
                        "         	page_url                                                      "+
                        "     ) o1                                                                "+
                        " ) o2                                                                    "+
                        " WHERE rn<=10                                                            "
        );





    }


}
